package com.tledu.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HdfsUtils {
    public static String readHdfsFile(String filePath, String encode) throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.newInstance(conf);
        Path path = new Path(filePath);
        FSDataInputStream inputStream = fs.open(path);
        // 读取文件流
        byte[] bytes = new byte[10240];
        int len = 0;
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        while ((len = inputStream.read(bytes)) > 0) {
            bos.write(bytes, 0, len);
        }
        // 获取最终的输出
        return bos.toString(encode);
    }

    /**
     * 实现的功能： 获取文件夹下的所有文件
     * 抽象出来输入和输出，关于递归一定找到出口，避免无限递归
     * 输入： 文件夹的地址
     * 输出： 文件集合
     * 出口：一个文件夹都是文件的时候，递归就结束了
     */
    public static List<String> getFileFromDir(String dirPath,FileSystem fs) throws IOException {
        Path path = new Path(dirPath);
        // 获取子目录
        FileStatus[] statuses = fs.listStatus(path);
        List<String> pathList = new ArrayList<>();
        for (FileStatus fileStatus : statuses) {
            if (fileStatus.isDirectory()) {
                // 文件夹
               List<String> subPathList = getFileFromDir(fileStatus.getPath().toString(), fs);
                pathList.addAll(subPathList);
            }else {
                // 文件
                pathList.add(fileStatus.getPath().toString());
            }
        }
        return pathList;
    }

    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.newInstance(conf);
        List<String> pathList = getFileFromDir("C:\\Users\\cyrus\\Desktop\\毕设公开课\\hdfs-job021\\data\\data", fs);
        pathList.forEach(item-> System.out.println(item));
    }

}
